from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.models.variable import Variable


class CustomOperator(BaseOperator):
    def execute(self, context):
        Variable.set(key="set_key", value="set_value", description="This is being set from task sdk")


with DAG("example_variable_set_sdk", schedule=None, catchup=False) as dag:
    CustomOperator(task_id="set_var")

from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk import Variable

Variable.set(key="set_key_top_level", value="set_value_top_level", description="This is being set from task sdk but at top level")

class CustomOperator(BaseOperator):
    def execute(self, context):
        print("Inside operator")


with DAG("example_variable_set_sdk_top_level", schedule=None, catchup=False) as dag:
    CustomOperator(task_id="set_var")

export AIRFLOW__WORKERS__SECRETS_BACKEND="airflow.secrets.local_filesystem.LocalFilesystemBackend"
export AIRFLOW__WORKERS__SECRETS_BACKEND_KWARGS='{"connections_file_path": "/files/conn.json", "variables_file_path": "/files/var.json"}'

{
    "var_a": "var_a_value"
}

root@765fc4803c7c:/opt/airflow# airflow config get-value workers secrets_backend
airflow.secrets.local_filesystem.LocalFilesystemBackend
root@765fc4803c7c:/opt/airflow# airflow config get-value workers secrets_backend_kwargs
{"connections_file_path": "/files/conn.json", "variables_file_path": "/files/var.json"

from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import Variable

def get_var_for_secrets_backend():
    variable = Variable.get("var_a")
    print("The variable from secrets backend is", variable)

def set_var_for_secrets_backend():
    Variable.set(key="var_a", value="trying to override")


with DAG("set_var_for_secrets_backend_dag", schedule=None, catchup=False) as dag:
    getter = PythonOperator(
        task_id="get_var_for_secrets_backend",
        python_callable=get_var_for_secrets_backend,
    )

    setter = PythonOperator(
        task_id="set_var_for_secrets_backend",
        python_callable=set_var_for_secrets_backend,
    )


    getter >> setter